package com.ht.api.kafka;

import java.util.ArrayList;
import java.util.List;

import org.springframework.context.annotation.Profile;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import com.ht.api.callback.CallbackManager;
import com.ht.api.db.Db;
import com.ht.api.db.DbConfig;
import com.ht.api.db.Entity;
import com.ht.api.db.StreamLoadConfig;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * kafka消费者
 * @author asq
 * @createTime 2024年12月7日 10:22:22
 */
@Slf4j
@Component
@Profile("prod")
public class KafkaConsumer {
	private static final long TENANTID = 1876286380045373440L;
	
	// 批量消费：回调事件
	@KafkaListener(topics = "callback", containerFactory = "kafkaListenerContainerFactory")
	public void consumeBatchByCallBack(List<String> messages, Acknowledgment ack) {
		log.info("[kafka]收到callback批量消息：size:{}条}", messages.size());

		// 处理消息
		messages.stream().forEach(m -> {
			try {
				// 解析消息中的data字段
				JSONObject message = JSONUtil.parseObj(m);
				String dataStr = message.getStr("data");
				
				// 解析data字段中的各顶具体数据
				JSONObject dataObj = JSONUtil.parseObj(dataStr);
				String appid = dataObj.getStr("appid");
				String data = dataObj.getStr("data");
				CallbackManager.callback(appid, data);
				
				log.info("[kafka]处理callback消息成功：{}", m);
			} catch (Exception e) {
				log.error(String.format("[kafka]处理callback消息失败：{}", e.getMessage()), e);
			}
		});

		// 手动提交偏移量
		ack.acknowledge();
	}
	
	// 批量消费：CDC数据
	@KafkaListener(topics = "CDC_" + TENANTID, containerFactory = "kafkaListenerContainerFactory")
	public void consumeBatchByCDC(List<String> messages, Acknowledgment ack) {
		log.info("[kafka]收到批量cdc消息：size:{}条}", messages.size());
		StreamLoadConfig streamLoadConfig = DbConfig.useTenantSLC(TENANTID);

		// 处理消息
		List<Entity> list = new ArrayList<>();
		messages.stream().forEach(m -> {
			try {
				// 解析消息中的data字段
				JSONObject message = JSONUtil.parseObj(m);
				message.remove("before");
				
				// 写入open_data
				String database = message.getStr("database");
				String table = message.getStr("table");
				String sourceCode = String.format("cdc_%s_%s_%s", TENANTID, database, table);
				Entity openData = Entity.create("open_data").set("source_code", sourceCode).set("data", message);
				list.add(openData);
				
				log.info("[kafka]解析cdc消息成功：{}", m);
			} catch (Exception e) {
				log.error(String.format("[kafka]处理cdc消息失败：{}", e.getMessage()), e);
			}
		});
		
		// 批量写入open_data
		Db.insertBatchWithStream(streamLoadConfig, list);
		log.info("[kafka]将cdc数据批量写入open_data成功，数据条数：{}", list.size());

		// 手动提交偏移量
		ack.acknowledge();
	}
}